package com.skyline.energy.executor.impl;

import java.lang.reflect.Array;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.skyline.energy.dataaccess.jdbc.JdbcDataAccessor;
import com.skyline.energy.dataaccess.jdbc.KeyHolder;
import com.skyline.energy.dataaccess.jdbc.RowMapper;
import com.skyline.energy.dataaccess.jdbc.ShardParam;
import com.skyline.energy.dataaccess.jdbc.SqlExecuteContext;
import com.skyline.energy.dataaccess.jdbc.SqlExecuteContextHolder;
import com.skyline.energy.dataaccess.jdbc.SqlType;
import com.skyline.energy.definition.JdbcBatchUpdateDefinition;
import com.skyline.energy.definition.JdbcDefinitionCollection;
import com.skyline.energy.definition.JdbcQueryDefinition;
import com.skyline.energy.definition.JdbcUpdateDefinition;
import com.skyline.energy.definition.ShardDefinition;
import com.skyline.energy.exception.DaoGenerateException;
import com.skyline.energy.exception.DataAccessException;
import com.skyline.energy.executor.DataAccessExecutor;
import com.skyline.energy.utils.CommonUtils;
import com.skyline.energy.utils.Page;

public class SqlExecutor implements DataAccessExecutor {
	private static final Log LOGGER = LogFactory.getLog(SqlExecutor.class);
	private final JdbcDataAccessor dataAccessor;
	private final Method method;
	private JdbcDefinitionCollection jdbcDefinitionCollection;

	public SqlExecutor(JdbcDataAccessor dataAccessor, Method method) throws DaoGenerateException {
		this.dataAccessor = dataAccessor;
		this.method = method;
		jdbcDefinitionCollection = new JdbcDefinitionCollection(method);
	}

	public JdbcDataAccessor getDataAccessor() {
		return dataAccessor;
	}

	public Method getMethod() {
		return method;
	}

	@Override
	public Object execute(Object obj, Object[] args) {
		if (!jdbcDefinitionCollection.needJdbcOpration()) {
			return null;
		}

		JdbcQueryDefinition queryDefinition = jdbcDefinitionCollection.getQueryDefinition();
		Object result = null;
		if (queryDefinition != null) {
			result = query(args);
		}

		JdbcUpdateDefinition updateDefinition = jdbcDefinitionCollection.getUpdateDefinition();
		if (updateDefinition != null) {
			result = update(args);
		}

		JdbcBatchUpdateDefinition batchUpdateDefinition = jdbcDefinitionCollection.getBatchUpdateDefinition();
		if (batchUpdateDefinition != null) {
			result = batchUpdate(args);
		}
		
		//clean SqlExecuteContext
		SqlExecuteContextHolder.setContext(null);
		
		return result;
	}

	private Object update(Object[] args) {
		JdbcUpdateDefinition updateDefinition = jdbcDefinitionCollection.getUpdateDefinition();
		Method[] getterMethods = updateDefinition.getGetterMethods();
		Integer[] parameterIndexes = updateDefinition.getParameterIndexes();
		boolean isReturnId = updateDefinition.isReturnId();

		String actualSql = updateDefinition.getActualSql(args);
		List<String> parameterNames = updateDefinition.getParsedSql().getParameterNames();
		Object[] paramArray = CommonUtils.fetchVlaues(getterMethods, parameterIndexes, args, parameterNames);

		initContext(updateDefinition, args, SqlType.UPDATE);

		JdbcDataAccessor dataAccessor = getDataAccessor();

		KeyHolder keyHolder = null;
		if (isReturnId) {
			keyHolder = dataAccessor.getKeyHolder();
		}

		LOGGER.info("Normal Update SQL:" + actualSql);
		int rows = dataAccessor.update(actualSql, keyHolder, paramArray);

		if (keyHolder != null) {
			return keyHolder.getKey();
		} else {
			return rows;
		}
	}

	private Object query(Object[] args) {
		JdbcQueryDefinition queryDefinition = jdbcDefinitionCollection.getQueryDefinition();
		Method[] getterMethods = queryDefinition.getGetterMethods();
		Integer[] parameterIndexes = queryDefinition.getParameterIndexes();
		RowMapper<?> rowMapper = queryDefinition.getRowMapper();
		int fetchSize = queryDefinition.getFetchSize();

		String actualSql = queryDefinition.getActualSql(args);
		List<String> parameterNames = queryDefinition.getParsedSql().getParameterNames();
		Object[] paramArray = CommonUtils.fetchVlaues(getterMethods, parameterIndexes, args, parameterNames);

		initContext(queryDefinition, args, SqlType.READ);

		int pageIndex = queryDefinition.getPageIndex();
		Page page = CommonUtils.getPageArgument(args, pageIndex);
		if (page != null) {
			return getDataAccessor().queryPage(actualSql, page, rowMapper, fetchSize, paramArray);
		}

		LOGGER.info("Query SQL:" + actualSql);

		List<?> result = getDataAccessor().query(actualSql, rowMapper, fetchSize, paramArray);
		if (queryDefinition.isUnique()) {
			if (result == null || result.isEmpty()) {
				return null;
			}
			if (result.size() > 1) {
				LOGGER.debug("result size > 1, fetch the 1st result");
			}
			return result.get(0);
		} else {
			return result;
		}
	}

	private Object batchUpdate(Object[] args) {
		JdbcBatchUpdateDefinition batchUpdateDefinition = jdbcDefinitionCollection.getBatchUpdateDefinition();
		String actualSql = batchUpdateDefinition.getActualSql(args);
		List<Object[]> paramArrays = generateBatchQueryArguments(args, batchUpdateDefinition);

		initContext(batchUpdateDefinition, args, SqlType.UPDATE);

		JdbcDataAccessor dataAccessor = getDataAccessor();

		KeyHolder keyHolder = null;
		if (batchUpdateDefinition.isReturnId()) {
			keyHolder = dataAccessor.getKeyHolder();
		}

		LOGGER.info("Normal Update SQL:" + actualSql);
		int[] rows = dataAccessor.batchUpdate(actualSql, paramArrays, keyHolder);

		if (keyHolder != null) {
			List<Number> keys = getKeysFromKeyHolder(keyHolder);
			return formatKeys(keys, batchUpdateDefinition);
		} else {
			return rows;
		}
	}

	private List<Object[]> generateBatchQueryArguments(Object[] args, JdbcBatchUpdateDefinition batchUpdateDefinition) {
		int batchSize = -1;
		Integer[] batchParamIndexes = batchUpdateDefinition.getBatchParamIndexes();
		for (int i = 0; i < batchParamIndexes.length; i++) {
			int index = batchParamIndexes[i];
			if (batchSize == -1) {
				batchSize = ArrayUtils.getLength(args[index]);
			} else {
				batchSize = Math.min(batchSize, ArrayUtils.getLength(args[index]));
			}
		}

		Method[] getterMethods = batchUpdateDefinition.getGetterMethods();
		Integer[] parameterIndexes = batchUpdateDefinition.getParameterIndexes();
		List<Object[]> paramArrays = new ArrayList<Object[]>(batchSize);
		List<String> parameterNames = batchUpdateDefinition.getParsedSql().getParameterNames();

		for (int i = 0; i < batchSize; i++) {
			Object[] cloneArgs = ArrayUtils.clone(args);

			for (int j = 0; j < batchParamIndexes.length; j++) {
				int index = batchParamIndexes[j];
				cloneArgs[index] = CommonUtils.fetchArrayValue(cloneArgs[index], i);
			}

			Object[] paramArray = CommonUtils.fetchVlaues(getterMethods, parameterIndexes, cloneArgs, parameterNames);
			paramArrays.add(paramArray);
		}

		return paramArrays;
	}

	private List<Number> getKeysFromKeyHolder(KeyHolder keyHolder) {
		List<Map<String, Object>> generatedKeys = keyHolder.getKeyList();
		int length = generatedKeys.size();

		List<Number> keys = new ArrayList<Number>();
		for (int i = 0; i < length; i++) {
			Iterator<Object> keyIter = generatedKeys.get(i).values().iterator();
			if (keyIter.hasNext()) {
				Object key = keyIter.next();
				if (!(key instanceof Number)) {
					String className = null;
					if (key != null) {
						className = key.getClass().getName();
					}
					throw new DataAccessException("The generated key is not of a supported numeric type. "
							+ "Unable to cast [" + className + "] to [" + Number.class.getName() + "]");
				}
				keys.add((Number) key);
			}
		}

		return keys;
	}

	private Object formatKeys(List<Number> keys, JdbcBatchUpdateDefinition batchUpdateDefinition) {
		if (batchUpdateDefinition.isReturnList()) {
			return keys;
		}

		int length = keys.size();

		Class<?> componentType = batchUpdateDefinition.getReturnComponentType();
		if (!CommonUtils.isTypePrimitive(componentType)) {
			Object array = Array.newInstance(componentType, length);
			System.arraycopy(keys.toArray(), 0, array, 0, length);
			return array;
		}

		return formatPrimitiveKeys(keys, componentType, length);

	}

	private Object formatPrimitiveKeys(List<Number> keys, Class<?> componentType, int length) {
		if (long.class.equals(componentType)) {
			Long[] array = keys.toArray(new Long[length]);
			return ArrayUtils.toPrimitive(array);
		}
		if (int.class.equals(componentType)) {
			Integer[] array = keys.toArray(new Integer[length]);
			return ArrayUtils.toPrimitive(array);
		}
		if (boolean.class.equals(componentType)) {
			Boolean[] array = keys.toArray(new Boolean[length]);
			return ArrayUtils.toPrimitive(array);
		}
		if (double.class.equals(componentType)) {
			Double[] array = keys.toArray(new Double[length]);
			return ArrayUtils.toPrimitive(array);
		}
		if (char.class.equals(componentType)) {
			Character[] array = keys.toArray(new Character[length]);
			return ArrayUtils.toPrimitive(array);
		}
		if (float.class.equals(componentType)) {
			Float[] array = keys.toArray(new Float[length]);
			return ArrayUtils.toPrimitive(array);
		}
		if (byte.class.equals(componentType)) {
			Byte[] array = keys.toArray(new Byte[length]);
			return ArrayUtils.toPrimitive(array);
		}
		if (short.class.equals(componentType)) {
			Short[] array = keys.toArray(new Short[length]);
			return ArrayUtils.toPrimitive(array);
		}
		throw new DataAccessException("can't parse return type of [" + componentType);
	}

	private void initContext(ShardDefinition definition, Object[] args, SqlType sqlType) {
		SqlExecuteContext context = new SqlExecuteContext();
		context.setSqlType(sqlType);
		
		Integer shardParamIndex = definition.getShardParamIndex();
		if(shardParamIndex != null) {
			Method shardGetter = definition.getShardGetter();
			String shardParamName = definition.getShardParamName();
			String shardTable = definition.getShardTable();
			Object value =CommonUtils.fetchVlaue(shardGetter, shardParamIndex, args, shardParamName);
			context.setShardParam(new ShardParam(shardTable, value));
		}
		
		SqlExecuteContextHolder.setContext(context);
	}
}
